-- https://blog.csdn.net/xianpanjia4616/article/details/116168585

# 开启binlog日志
log-bin=mysql-bin
binlog_format=ROW
server_id=142

CREATE TABLE tbl_data_abc (
    id bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增id',
    name varchar(63) NOT NULL DEFAULT '' COMMENT '姓名',
    age int NOT NULL DEFAULT '0' COMMENT '年龄',
    city varchar(63) NOT NULL DEFAULT '' COMMENT '所在城市',
    phone varchar(63) NOT NULL DEFAULT '' COMMENT '移动电话',
    action tinyint(4) NOT NULL DEFAULT '0' COMMENT '记录状态',
    action_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录修改时间',
    create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uniq_name` (`name`),
    KEY `idx_action` (`action`),
    KEY `idx_action_time` (`action_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='测试flinkSQL4mysqlCdc';

CREATE TABLE mysql_cdc_tbl_data_abc (
    id bigint,
    name string,
    age int,
    city string,
    phone string,
    action tinyint,
    action_time string,
    create_time string
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = 'lixy0302',
  'database-name' = 'db58_zp_bi',
  'table-name' = 'tbl_data_abc',
  'debezium.snapshot.mode' = 'initial'
)

CREATE TABLE ods_zp_n_tbl_data_abc (
    id bigint,
    name string,
    age int,
    city string,
    phone string,
    action tinyint,
    action_time string,
    create_time string
) WITH (
'connector' = 'kafka',
'topic' = 'ods_zp_n_tbl_data_abc',
'scan.startup.mode' = 'earliest-offset',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'debezium-json'
)

insert into ods_zp_n_tbl_data_abc
select *
from mysql_cdc_tbl_data_abc




INSERT INTO tbl_data_abc(name,age,city,phone) VALUES ('hadoop',00,'shanghai',778899);
UPDATE tbl_data_abc set age = '00' WHERE name = 'JasonLee';
INSERT INTO tbl_data_abc(name,age,city,phone) VALUES ('clickhouse',00,'shanghai',778899);
UPDATE tbl_data_abc set age = '20' WHERE name = 'clickhouse';




-- 设置cp路径时间间隔
SET 'state.checkpoint.path' = 'file:///tmp/flink/checkpoint';
-- 设置cp存储类型
set 'state.checkpoint-storage' ='filesystem'
-- 设置cp时间间隔
SET 'execution.checkpointing.interval' = '3s';


drop table if EXISTS ods_dataGen_order;

-- 创建一个订单表，用datagen生成数据
create table if not EXISTS ods_dataGen_order(
	id_ int ,
	price decimal(8,3) ,
	product_id int,
	user_id int,
	order_date TIMESTAMP ,
	PRIMARY KEY(id_) NOT ENFORCED
)WITH (
 'connector' = 'datagen',
 'rows-per-second'='3',
 'number-of-rows'='1000',
 'fields.id_.kind'='sequence',
 'fields.id_.start'='1',
 'fields.id_.end'='1000',

 'fields.price.kind'='random',
 'fields.price.min'='10',
 'fields.price.max'='1000',

 'fields.product_id.kind'='random',
 'fields.product_id.min'='1',
 'fields.product_id.max'='10',

 'fields.user_id.kind'='random',
 'fields.user_id.min'='1',
 'fields.user_id.max'='50'
);

drop table if EXISTS dim_product_info;

create table if not EXISTS dim_product_info(
    id_ int ,
    product_id int,
    product_name string,
    PRIMARY KEY(id_) NOT ENFORCED
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = '192.168.45.1',
 'port' = '3306',
 'username' = 'root',
 'password' = '123456',
 'database-name' = 'test',
 'table-name' = 'product_info'

 );


-- drop table if EXISTS ads_restul;
-- create table if not EXISTS ads_restul(
--     id_ int ,
-- 	price decimal(8,3) ,
-- 	product_id int,
-- 	user_id int,
-- 	order_date TIMESTAMP ,
-- 	id_2 int ,
--     product_id_2 int,
--     product_name string,
--     PRIMARY KEY(id_) NOT ENFORCED
-- ) WITH ('connector' = 'print');


-- INSERT INTO  ads_restul
-- 实时ETL ，订单流水关联产品档
select  a.id_  , a.price  , a.product_id , a.user_id , a.order_date  , b.id_  , b.product_id , b.product_name
 from ods_dataGen_order a,dim_product_info b where a.product_id=b.product_id;